package com.examples.mysql2mysql.stream;

import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.*;

public class Flink2JdbcReader extends
	RichSourceFunction<JSONObject> {
	private static final long serialVersionUID = 3334654984018091675L;

	private Connection connect = null;
	private PreparedStatement ps = null;
	private PreparedStatement updps = null;

	private boolean running=false;

	/*
	 * (non-Javadoc)
	 *
	 * @see org.apache.flink.api.common.functions.AbstractRichFunction#open(org.
	 * apache.flink.configuration.Configuration) to use open database connect
	 */
	@Override
	public void open(Configuration parameters) throws Exception {
		super.open(parameters);
		running =true;
		Class.forName("com.mysql.jdbc.Driver");
		connect = (Connection) DriverManager.getConnection("jdbc:mysql://127.0.0.1:3306/f_saas_cfg?useUnicode=true&characterEncoding=UTF-8&useAffectedRows=true&allowMultiQueries=true&useSSL=false&zeroDateTimeBehavior=convertToNull&serverTimezone=GMT%2B8", "root", "123456");
		ps = (PreparedStatement) connect
			.prepareStatement("select * from test_table a where a.del ='0'");
	}

	@Override
	public void run(SourceContext<JSONObject> sourceContext) throws Exception {
		while(running){
			ResultSet resultSet = ps.executeQuery();
			while (resultSet.next()) {
				JSONObject jsonObject = resultSetToJsonObject(resultSet);
				// 先将数据删除，然后在后续调用
				updps = (PreparedStatement) connect
					.prepareStatement("update  test_table set del = '1' where id = ?");
				updps.setString(1,jsonObject.getString("id"));
				int vv = updps.executeUpdate();
				System.out.println("本次修改"+vv+"条数据");


				// 将数据备份到 01 表
				// INSERT INTO users (username, email, password)
				// VALUES ('JohnDoe', 'john@example.com', 'hashed_password');
				updps = (PreparedStatement) connect
						.prepareStatement("INSERT INTO test_table01(`id`, `name`, `del`) VALUES (?, ?, ?);");
				updps.setString(1, jsonObject.getString("id"));
				updps.setString(2, jsonObject.getString("name"));
				updps.setString(3, jsonObject.getString("del"));
				updps.execute();
				System.out.println("本次修改"+vv+"条数据");
				sourceContext.collect(jsonObject);
			}
		}
	}

	public static JSONObject resultSetToJsonObject(ResultSet rs) throws SQLException, JSONException {
		// json对象
		JSONObject jsonObj = new JSONObject();
		// 获取列数
		ResultSetMetaData metaData = rs.getMetaData();
		int columnCount = metaData.getColumnCount();
		// 遍历ResultSet中的每条数据
		// 遍历每一列
		for (int i = 1; i <= columnCount; i++) {
			String columnName =metaData.getColumnLabel(i);
			String value = rs.getString(columnName);
			jsonObj.put(columnName, value);
		}
		return jsonObj;

	}

	/*
	 * (non-Javadoc)
	 *
	 * @see
	 * org.apache.flink.streaming.api.functions.source.SourceFunction#cancel()
	 * colse database connect
	 */
	@Override
	public void cancel() {
		running =false;
		try {
			super.close();
			if (connect != null) {
				connect.close();
			}
			if (ps != null) {
				ps.close();
			}
		} catch (Exception e) {
			e.printStackTrace();
		}

	}

}
